package com.beta.mq.redis.consumer.work;

import com.alibaba.fastjson.JSON;
import com.beta.mq.redis.config.RefreshBean;
import com.beta.mq.redis.consumer.Consumer;
import com.beta.mq.redis.consumer.annotation.OnBatchMessage;
import com.beta.mq.redis.consumer.annotation.OnMessage;
import com.beta.mq.redis.model.Message;
import com.beta.mq.redis.producer.annotation.ToQueue;

import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.util.CollectionUtils;

import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * 阻塞队列式消费策略
 */
@Slf4j
public class RedisBlockingQueueListener implements InitializingBean {
    Logger log = LoggerFactory.getLogger(this.getClass());

    @Autowired
    private ApplicationContext applicationContext;
    @Autowired
    private Consumer consumer;
    @Autowired
    private RefreshBean refreshBean;

    private static final List<Thread> pools = new ArrayList<>();

    public void listen() {
        //把所有加了注解的类及方法找出来，进行消息处理！
        //获得所有消费者Bean
        Map<String, Object> beans = applicationContext.getBeansWithAnnotation(com.beta.mq.redis.consumer.annotation.Consumer.class);
        //获得Bean Factory
        for (Map.Entry<String, Object> entry : beans.entrySet()) {
            String name = entry.getKey();
            Object bean = entry.getValue();
            Class clazz = applicationContext.getType(name);
            for (Method method : clazz.getMethods()) {
                //启动线程的数量  取决于消费者的数量
                if (method.isAnnotationPresent(OnMessage.class)) {
                    OnMessage onMessage = method.getAnnotation(OnMessage.class);
                    putConsumerPool(onMessage.topic(), method, bean);
                }
                //
                if (method.isAnnotationPresent(OnBatchMessage.class)) {
                    OnBatchMessage config = method.getAnnotation(OnBatchMessage.class);
                    putConsumerPoolForBatch(config, method, bean);
                }
            }
        }
    }

    private void putConsumerPool(String topic, Method method, Object bean) {
        Thread thread = new Thread(() -> {
            while (true) {
                try {
                    //获取消息
                    Message message;
                    while ((message = consumer.getMessage(topic)) != null) {
                        if (message.getExpireAt() != ToQueue.ExpireTime.NEVER_EXPIRES && message.getExpireAt() < System.currentTimeMillis()) {
                            //说明这是一个过期任务，记录日志后丢弃掉
                            continue;
                        }

                        if(refreshBean.getSingelConsumerSwitch() == 0){
                            try {
                                method.invoke(bean, message);
                            } catch (Exception e) {
                                //如果消息执行失败，重试
                                consumer.retry(topic, message);
                            }
                        }else{
                            log.warn("消息丢弃，{}", JSON.toJSONString(message));
                        }

                    }
                } catch (Exception e) {
                    log.warn("消息失败！{}", e.getMessage(), e);
                }
                //消费完后停顿一点时间
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException ignore) {
                }
            }
        });
        thread.setName("ConsumeWorker-" + topic);
        thread.start();
        pools.add(thread);
    }

    private void putConsumerPoolForBatch(OnBatchMessage config, Method method, Object bean) {
        Thread thread = new Thread(() -> {
            while (true) {
                try {
                    //获取消息
                    List<Message> messages = consumer.getBatchMessage(config.topic(), config.limit());
                    if (!CollectionUtils.isEmpty(messages)) {
                        messages = messages.stream()
                                .filter(msg -> msg.getExpireAt() == ToQueue.ExpireTime.NEVER_EXPIRES || msg.getExpireAt() >= System.currentTimeMillis())
                                .collect(Collectors.toList());

                        try {
                            method.invoke(bean, messages);
                        } catch (Exception e) {
                            //如果消息执行失败，重试
                            consumer.retryBatch(config.topic(), messages);
                        }

                    }
                } catch (Exception e) {
                    log.warn("消息失败！{}", e.getMessage(), e);
                }
                //消费完后停顿一点时间
                try {
                    Thread.sleep(config.interval());
                } catch (InterruptedException ignore) {
                }
            }
        });
        thread.setName("BatchConsumeWorker-" + config.topic());
        thread.start();
        pools.add(thread);
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        listen();
    }
}
